package com.gitee.ixtf.vertx.rabbitmq.guice

import com.google.inject.AbstractModule
import com.google.inject.Provides
import com.rabbitmq.client.Address
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.impl.nio.NioParams
import io.vertx.rabbitmq.RabbitMQOptions
import reactor.core.scheduler.Schedulers
import reactor.rabbitmq.*
import reactor.rabbitmq.Utils.singleConnectionMono

abstract class RabbitMQGuice : AbstractModule() {
  override fun configure() {
    bind(RabbitMQOptions::class.java).toProvider {
      RabbitMQOptions().also { options ->
        options.setAutomaticRecoveryEnabled(true)
        prepare(options)
      }
    }
  }

  protected abstract fun prepare(options: RabbitMQOptions)

  @Provides
  private fun ConnectionFactory(options: RabbitMQOptions): ConnectionFactory {
    val connectionFactory = ConnectionFactory()
    connectionFactory.useNio()
    val nioParams = NioParams().setWriteQueueCapacity(10000000).setWriteEnqueuingTimeoutInMs(60000)
    connectionFactory.nioParams = nioParams
    connectionFactory.host = options.host
    connectionFactory.username = options.user
    connectionFactory.password = options.password
    return connectionFactory
  }

  @Provides
  private fun Sender(connectionFactory: ConnectionFactory): Sender {
    val connectionMono = singleConnectionMono {
      val address = Address(connectionFactory.host)
      connectionFactory.newConnection(arrayOf(address))
    }
    val channelPoolOptions = ChannelPoolOptions().maxCacheSize(10)
    val senderOptions =
        SenderOptions()
            .connectionFactory(connectionFactory)
            .connectionMono(connectionMono)
            .resourceManagementScheduler(Schedulers.boundedElastic())
            .channelPool(ChannelPoolFactory.createChannelPool(connectionMono, channelPoolOptions))
    return RabbitFlux.createSender(senderOptions)
  }
}
